package org.yuanzheng.transformation

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.yuanzheng.source.{CustomSource, StationLog}

/**
 * @author yuanzheng
 * @date 2020/6/17-20:44
 */
object TestSplitAndSelect {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val stream: DataStream[StationLog] = streamEnv.addSource(new CustomSource)
    //使用split和select
    val splitStream: SplitStream[StationLog] = stream.split(log => {
      if (log.callType == "success") Seq("ok") else Seq("error")
    })
    val stream1: DataStream[StationLog] = splitStream.select("ok")
    val stream2: DataStream[StationLog] = splitStream.select("error")
    stream1.print("通话成功")
    stream2.print("通话失败")
    streamEnv.execute()
  }
}
